package com.study.flink.java.day04_async.async_io.http;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.study.flink.java.day04_async.ActivityBean;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpResponse;
import org.apache.http.NameValuePair;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Supplier;

public class BalanceAsyncToActivityBeanFunction extends RichAsyncFunction<String, ActivityBean> {

    // 外部接口地址
    private transient static final String requestUrl = "http://yan.taobiso.com/Handler/Api.ashx?type=GetBalance";
    private transient static final String UID = "225881";

    private transient CloseableHttpAsyncClient httpAsyncClient = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
         // 初始化异步的HttpClient
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(3000)
                .setConnectTimeout(3000)
                .build();
        httpAsyncClient = HttpAsyncClients.custom()
                .setMaxConnTotal(20)
                .setDefaultRequestConfig(requestConfig).build();
        httpAsyncClient.start();

    }

    @Override
    public void asyncInvoke(String line, ResultFuture<ActivityBean> resultFuture) throws Exception{
        System.out.println(line);
        // line example: u001,A1,2020-02-26 10:11:12,1,北京市
        // line example: u002,A2,2020-02-25 10:11:12,1,厦门市
        // line example: u003,A3,2020-02-25 20:44:55,1,泉州市
        // line example: u003,A4,2020-02-23 08:28:45,2,莆田市
        String[] fields = line.split(",");
        String uid = fields[0];
        String aid = fields[1];
        String time = fields[2];
        int eventType = Integer.parseInt(fields[3]);
        String province = fields[4];

        //外部接口查询实时余额
        String ticks = DateUtil.format(new Date(System.currentTimeMillis()), DatePattern.NORM_DATETIME_PATTERN);
        String token = getToken(ticks);
        Map<String, String> paramMap = new HashMap<>();
        paramMap.put("uid", UID);
        paramMap.put("ticks", ticks);
        paramMap.put("nick", "");
        paramMap.put("token", token);

        HttpPost httpPost = new HttpPost(requestUrl);
        List<NameValuePair> content = new ArrayList<>();
        Iterator iterator = paramMap.entrySet().iterator();           //将content生成entity
        while(iterator.hasNext()){
            Map.Entry<String,String> elem = (Map.Entry<String, String>) iterator.next();
            content.add(new BasicNameValuePair(elem.getKey(), elem.getValue()));
        }
        if(content.size() > 0) {
            UrlEncodedFormEntity entity = new UrlEncodedFormEntity(content, "UTF-8");
            httpPost.setEntity(entity);
        }
        // 使用异步的httpAsyncClient，查外部别人定义的API
        Future<HttpResponse> future = httpAsyncClient.execute(httpPost, null);
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    HttpResponse response = future.get();
                    String balance = null;
                    if(response.getStatusLine().getStatusCode() == 200) {
                        /**
                         * {
                         *   "Balance": "273.70",
                         *   "success": true,
                         *   "info": ""
                         * }
                         */
                        String result = EntityUtils.toString(response.getEntity());
                        System.out.println(result);
                        JSONObject jsonObject = JSONObject.parseObject(result);
                        balance = jsonObject.getString("Balance");
                    }
                    return balance;
                } catch (Exception e) {
                    return null;
                }
            }
        }).thenAccept((String balance) -> {
           resultFuture.complete(Collections.singleton(ActivityBean.of(uid, aid, "", time, eventType, province, balance)));
        });
    }

    @Override
    public void close() throws Exception {
        super.close();
        httpAsyncClient.close();
    }

    private String getToken(String ticks) {
        StringBuffer query = new StringBuffer();
        query.append(UID);
        query.append(ticks);
        query.append("13C0D11FDC5A4EBBA3F5");
        String param = new String(query);
        byte[] data2 = null;
        try {
            data2 = param.getBytes("UTF-8");
        } catch (Exception e) {
        }
        String sign = DigestUtils.md5Hex(data2);
        return sign;
    }

}
